package join;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.lib.MultipleInputs;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class ClickUserInnerJoinAuction extends Configured implements Tool {

	private static final String auctionSeparator = "\u0001";
	private static final String clickUserSeparator = "\t";

	public static class AuctionMap extends MapReduceBase implements
			Mapper<LongWritable, Text, Text, Text> {
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			String line = value.toString();
			String[] auction = line.split(auctionSeparator, 34);
			if (auction.length == 34 && isValid(auction[18]))
				output.collect(new Text(auction[0]), new Text(auction[18]));
		}
	}

	public static class ClickUserMap extends MapReduceBase implements
			Mapper<LongWritable, Text, Text, Text> {
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			String line = value.toString();
			String[] clickUser = line.split(clickUserSeparator, 29);
			if (clickUser.length == 29) {
				if (isValid(clickUser[19]))
					output.collect(new Text(clickUser[19]), new Text(line));
			}
		}
	}

	public static class Reduce extends MapReduceBase implements
			Reducer<Text, Text, Text, Text> {

		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			ArrayList<String> container = new ArrayList<String>();
			String auctionDigitalId = "";
			while (values.hasNext()) {
				String result = values.next().toString();
				if (!result.contains(clickUserSeparator)) {
					auctionDigitalId = result;
				} else {
					container.add(result);
				}
			}
			if (isValid(auctionDigitalId)) {
				for (String s : container) {
					output.collect(new Text(auctionDigitalId), new Text(s));
				}
			}
		}
	}

	public static boolean isValid(String src) {
		return (src != null && !src.equals(""));
	}

	public int run(String[] args) throws Exception {
		String date = args[0];
		// ClickUser
		String clickUserPathStr = "/group/tbsc-dev/xzwang/cyliu/join/intermediate/" + date + "/clickUser";
		Path clickUserPath = new Path(clickUserPathStr);
		// Auction
		String auctionPathStr = "/group/taobao/taobao/hive/r_auction_auctions/pt=20100104000000/if_online=1";
		String auctionPathB90Str = "/group/taobao/taobao/hive/r_auction_auctions/pt=20100104000000/if_online=0";
		Path auctionPath = new Path(auctionPathStr);
		Path auctionPathB90 = new Path(auctionPathB90Str);
		
		// Output
		String clickUserAuctionPathStr = "/group/tbsc-dev/xzwang/cyliu/join/intermediate/" + date + "/clickUserAuction";
		Path outputPath = new Path(clickUserAuctionPathStr);
		JobConf conf = new JobConf(getConf(), ClickUserInnerJoinAuction.class);
		conf.setJobName("ClickUserInnerJoinAuction");
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);
		MultipleInputs.addInputPath(conf, auctionPath, TextInputFormat.class,
				AuctionMap.class);
		MultipleInputs.addInputPath(conf, auctionPathB90, TextInputFormat.class,
				AuctionMap.class);
		MultipleInputs.addInputPath(conf, clickUserPath, TextInputFormat.class,
				ClickUserMap.class);
		FileOutputFormat.setOutputPath(conf, outputPath);
		conf.setReducerClass(Reduce.class);
		JobClient.runJob(conf);
		return 0;
	}

	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(),
				new ClickUserInnerJoinAuction(), args);
		System.exit(res);
	}

}
